跨端数据格式统一与联调测试

关联：索引

要解决的问题

• Web 前端、rosbridge、ROS2 节点都能通信，为什么一联调就出现“字段不一致、类型漂移、状态解释冲突”？
• 同一个业务事件（如“机械臂执行结果”），为什么在前端是 ok，在后端变成 success，在 ROS2 又变成 result？
• 异常数据（字段缺失、时间戳单位混乱、字符串数字）如何做到“不中断主流程 + 可追溯”？
• 端到端联调如何从“能跑通”升级到“可复现、可定位、可回归”？
• 场景化测试用例如何设计，才能覆盖真实项目里的高频故障点？

章节内容（本讲核心）：

• 统一消息结构体（Envelope + Payload）
• 字段约定（命名、类型、必填/选填、时间戳与追踪字段）
• 异常数据兼容（容错解析、降级策略、告警与日志）
• 端到端联调流程（Web ↔ FastAPI ↔ ROS2）
• 场景化测试用例设计（功能、异常、性能、回归）

与前置知识衔接（避免重复）：

• 已学：WebSocket 生命周期、消息收发、状态管理
• 已学：FastAPI WebSocket 服务端搭建、Vue3 客户端封装
• 已学：rosbridge 连接 ROS2、订阅与指令下发
• 本讲不重复：环境安装、基础连接命令、组件从零搭建
• 本讲定位：在“已连通”的基础上，解决“格式不统一导致的联调成本高”问题，建立统一协议与测试闭环

项目工坊：

• 统一前后端、Web-ROS2 数据格式并全链路联调

项目选择建议（本讲统一）

• 推荐：使用综合项目 10_cross_end_e2e_lab/（第 10 阶段综合实战项目）同时覆盖“数据面（实时推送/展示）+ 控制面（指令下发/状态回传）”，更贴合“跨端数据格式统一与联调测试”的主题。

学生任务：

• 完成联调
• 记录日志
• 修复格式问题

大模型任务：

• AI 生成统一通信协议文档（可直接纳入项目仓库）

作业：

统一协议草案（本讲统一口径）

1. 顶层 Envelope（跨端统一）

{
  "schema_version": "1.0.0",
  "trace_id": "T-20260415-a1b2c3",
  "msg_id": "M-20260415-0001",
  "source": "web",
  "target": "ros2",
  "topic": "/sorting_arm/cmd",
  "event": "arm.command.request",
  "ts_ms": 1776200000123,
  "content_type": "application/json",
  "payload": {
    "device_id": "arm_01",
    "action": "stop",
    "params": {
      "reason": "user_click"
    }
  }
}

• schema_version：协议版本。版本升级时先兼容旧版本，再逐步迁移，避免一次性断联。
• trace_id：整条链路追踪 ID。一次业务动作（请求→执行→回执）建议复用同一个 trace_id。
• msg_id：单条消息唯一 ID。用于幂等、去重和日志定位。
• source/target：消息来源与目标，便于排查“是谁发错了字段”。
• topic/event：一个偏通信路由（topic），一个偏业务语义（event），避免只靠字符串猜用途。
• payload：业务体，不同场景可变；Envelope 字段保持稳定。

2. 错误回执统一格式（建议）

{
  "schema_version": "1.0.0",
  "trace_id": "T-20260415-a1b2c3",
  "msg_id": "M-20260415-0002",
  "source": "ros2",
  "target": "web",
  "topic": "/sorting_arm/status",
  "event": "arm.command.reject",
  "ts_ms": 1776200000456,
  "content_type": "application/json",
  "payload": {
    "ok": false,
    "code": "BAD_FIELD_TYPE",
    "message": "params.speed should be number",
    "field": "payload.params.speed"
  }
}

• ok/code/message/field：最小错误四元组，既能给页面提示，也能给开发定位。
• trace_id 回传：保证请求和错误可一键关联，避免“只看到错，不知道哪条请求导致的”。

1. 联调失败最多的不是“连不上”，而是“字段语义不一致”。
2. 没有统一 Envelope，日志很难跨端串起来。
3. 没有错误回执规范，学生只能看到“失败”，看不到“为什么失败”。

1) 命名约定（本课程统一）

• JSON 字段统一使用 snake_case（与 ROS2/Python 更一致）。
• 布尔字段必须使用肯定语义，如 ok、is_online，避免双重否定。
• 时间统一 ts_ms（毫秒），不混用 timestamp、time、ts。
• ID 字段统一后缀 _id，如 trace_id、msg_id、device_id。

2) 必填与选填建议

• 必填：schema_version、trace_id、msg_id、event、ts_ms、payload
• 选填：target、content_type、topic（某些纯业务事件可不依赖 topic）

3) 类型约束建议

• ts_ms 必须是 number 且有限值（禁止字符串时间戳）。
• trace_id/msg_id 必须是非空字符串。
• payload 必须是 object（禁止数组/纯字符串作为顶层业务体）。

type Envelope<TPayload extends Record<string, unknown>> = {
  schema_version: string
  trace_id: string
  msg_id: string
  source: string
  target?: string
  topic?: string
  event: string
  ts_ms: number
  content_type?: 'application/json'
  payload: TPayload
}

type ValidateResult<T> =
  | { ok: true; data: T }
  | { ok: false; code: string; message: string; field?: string }

function isObject(x: unknown): x is Record<string, unknown> {
  return typeof x === 'object' && x !== null && !Array.isArray(x)
}

export function validateEnvelope(input: unknown): ValidateResult<Envelope<Record<string, unknown>>> {
  if (!isObject(input)) {
    return { ok: false, code: 'BAD_BODY', message: 'body must be object' }
  }

  const requiredStringFields = ['schema_version', 'trace_id', 'msg_id', 'source', 'event'] as const
  for (const field of requiredStringFields) {
    if (typeof input[field] !== 'string' || input[field] === '') {
      return { ok: false, code: 'BAD_FIELD', message: `${field} must be non-empty string`, field }
    }
  }

  if (typeof input.ts_ms !== 'number' || !Number.isFinite(input.ts_ms)) {
    return { ok: false, code: 'BAD_FIELD_TYPE', message: 'ts_ms must be finite number', field: 'ts_ms' }
  }

  if (!isObject(input.payload)) {
    return { ok: false, code: 'BAD_FIELD_TYPE', message: 'payload must be object', field: 'payload' }
  }

  return { ok: true, data: input as Envelope<Record<string, unknown>> }
}

• Envelope<TPayload>：把“稳定外壳”和“可变业务体”分离，后续不同业务只改 payload 类型。
• ValidateResult：统一校验返回结构，前端页面与后端接口都能复用同样的错误口径。
• isObject()：先做基础收敛，防止直接访问 unknown 字段导致运行时异常。
• requiredStringFields 循环：避免手写一堆重复 if，同时保证错误信息格式一致。
• Number.isFinite(input.ts_ms)：拦截 NaN/Infinity，这是联调中常见脏值。

from pydantic import BaseModel, Field
from typing import Dict, Any, Optional, Literal

class EnvelopeModel(BaseModel):
    schema_version: str = Field(min_length=1)
    trace_id: str = Field(min_length=1)
    msg_id: str = Field(min_length=1)
    source: str = Field(min_length=1)
    target: Optional[str] = None
    topic: Optional[str] = None
    event: str = Field(min_length=1)
    ts_ms: int
    content_type: Optional[Literal["application/json"]] = None
    payload: Dict[str, Any]

• Field(min_length=1)：约束字符串非空，避免“字段存在但无效”。
• Optional：只对确实可选字段使用，减少协议歧义。
• payload: Dict[str, Any]：先保证结构合法，业务细分字段可在下一层模型继续校验。
• Literal["application/json"]：把可选字段值域固定住，避免拼写漂移。

• 影响追踪、安全、幂等的字段异常，一律拒绝。
• 不影响主链路的可选字段，可降级并打 warning 日志。

六、练习（至少 2 题）

1. 把 validateEnvelope 扩展为：ts_ms 允许秒级时间戳输入并自动转毫秒。
   提示：先判断值域再转换，转换后仍要保证有限数。
2. 设计一个 event 命名规范（例如 domain.action.stage），并给出 3 个本项目示例。
   提示：至少覆盖请求、成功回执、失败回执三类。

八、学生任务（提交物与标准）

• 提交物：协议草案（Markdown）+ 校验代码片段 + 3 条测试消息（1 成功、2 失败）

1. 链路连通性检查（连接是否建立）
2. 协议合法性检查（Envelope 是否通过）
3. 业务语义检查（event/payload 是否符合预期）
4. 异常回执检查（错误是否可定位）
5. 回归复测（修复后是否破坏其他场景）

1) 启动联调所需进程（本讲配套项目：直连 rosbridge）

# 终端 A：启动 rosbridge（示例）
ros2 launch rosbridge_server rosbridge_websocket_launch.xml port:=9090 address:=0.0.0.0

• address:=0.0.0.0：多机联调时可访问；若只本机联调建议改成 localhost。

# 终端 B：启动 ROS2 模拟节点（示例）
cd 10_cross_end_e2e_lab/ros2_ws
colcon build --symlink-install
source install/setup.bash
ros2 run sorting_arm_mock arm_mock

• 该 mock 同时发布 /device/state_json、/device/params_json、/vision/detections_json，用于实时监控页面联调。

# 终端 C：启动 Web 项目（本讲综合项目）
cd 10_cross_end_e2e_lab
npm.cmd install
npm.cmd run dev

• npm.cmd install：安装依赖（PowerShell 执行策略限制时推荐用 npm.cmd）。
• npm.cmd run dev：启动开发服务器，浏览器打开终端输出地址。
• 项目内有两个页面入口：实时监控 / 指令下发；两页共用同一个 rosbridge URL 输入（默认 ws://localhost:9090，也可改为端口映射 ws://localhost:19090）。

1.1) 可选：引入 FastAPI 网关（扩展路线）

推荐启动方式（示例）：

# 在你的 FastAPI 项目目录执行（示例）
fastapi dev

• fastapi dev：开发模式启动（自动 reload），需要在 FastAPI 项目中配置 entrypoint。

2) 联调时的最小消息流（Web 侧示例）

{
  "schema_version": "1.0.0",
  "trace_id": "T-20260415-001",
  "msg_id": "M-req-001",
  "source": "web",
  "target": "ros2",
  "topic": "/sorting_arm/cmd",
  "event": "arm.command.request",
  "ts_ms": 1776200100123,
  "payload": {
    "device_id": "arm_01",
    "action": "home",
    "params": {}
  }
}

• 这条消息用于验证“请求路径”。

建议每条日志至少包含：

• trace_id
• msg_id
• event
• stage（如 web.send / ros2.validate / ros2.execute / ros2.reply；若启用网关可增加 gateway.*）
• result（ok/fail）
• error_code（失败时）

示例日志（JSON line）：

{"trace_id":"T-20260415-001","msg_id":"M-req-001","event":"arm.command.request","stage":"ros2.validate","result":"fail","error_code":"BAD_FIELD_TYPE","field":"payload.params.speed"}

• 一行一事件，便于 grep/聚合统计。
• 出错时必须带 field，减少“猜字段”的时间。

2.1) ROS2 侧联调自检（推荐按顺序执行）

目标：在 ROS2 侧用最少命令确认“消息已到达 + Envelope 合法 + trace_id 能串联 + 回执能匹配”。

1. 确认 rosbridge 暴露了目标话题（类型与 QoS 也能看到）：

ros2 topic info -v /sorting_arm/cmd
ros2 topic info -v /sorting_arm/status

• ros2 topic info -v：显示话题类型与 QoS，便于排查“订阅成功但收不到”的 QoS/类型问题。

2. 抓取一条 /sorting_arm/cmd 并检查 data 字段是否是 Envelope JSON：

ros2 topic echo /sorting_arm/cmd --once

• --once：只抓取 1 条消息，避免刷屏影响观察。
• 你应该看到 data: '{"schema_version":"1.0.0",...,"payload":{...}}' 这种“JSON 字符串”。

3. 用一键校验脚本验证 Envelope 的必填字段（推荐复制执行）：

ros2 topic echo /sorting_arm/cmd --once | python3 - << 'PY'
import json, re, sys

text = sys.stdin.read()
m = re.search(r"data:\\s*'(?P<j>.*)'\\s*$", text, re.S)
if not m:
    print("FAIL: cannot find std_msgs/String.data")
    raise SystemExit(1)

raw = m.group("j").replace("\\\\'", "'")
obj = json.loads(raw)
need = ["schema_version","trace_id","msg_id","source","event","ts_ms","payload"]
miss = [k for k in need if k not in obj]
if miss:
    print("FAIL: missing fields:", miss)
    raise SystemExit(2)
if not isinstance(obj["payload"], dict):
    print("FAIL: payload must be object")
    raise SystemExit(3)
print("OK: trace_id =", obj["trace_id"], "event =", obj["event"])
PY

• 这段脚本做的事：从 ros2 topic echo 输出中提取 data 字符串 → json.loads → 检查必填字段。
• 如果校验失败：优先回到 Web 端检查是否按统一协议发送（字段命名/类型/是否漏包 Envelope）。

4. 抓取一条 /sorting_arm/status，核对两件事：

ros2 topic echo /sorting_arm/status --once

• data 必须同样是 Envelope JSON 字符串。
• payload.last_cmd_id 必须能与 Web 端发出的 payload.cmd_id 对上（这是“闭环匹配”的关键证据）。

常见错误快速定位：

• /sorting_arm/cmd 有消息但 /sorting_arm/status 没消息：优先检查 ROS2 mock 是否已运行、是否订阅了 /sorting_arm/cmd。
• data 不是 JSON 字符串：检查发布端是否把 JSON 放进了 std_msgs/msg/String.data（而不是把对象直接塞进 msg）。
• 有 Envelope 但字段缺失：直接用上面的“一键校验脚本”看缺哪一项，再回到发送端修复。

四、练习（至少 2 题）

1. 设计“时间戳异常”测试：秒/毫秒/微秒混入时，系统如何识别并处理？
   提示：先定义判定阈值，再定义修复策略与拒绝策略。
2. 设计“部分链路失败”测试：网关校验通过但 ROS2 执行失败，前端如何展示状态？
   提示：区分 validate_fail 与 execute_fail 两类错误码。

• 提交物：联调记录表（时间、用例、结果、日志截图）、修复清单、回归结果

课后作业

参考与延伸

• ROS2 文档：https://docs.ros.org/
• rosbridge_suite：https://github.com/RobotWebTools/rosbridge_suite
• FastAPI 文档：https://fastapi.tiangolo.com/
• JSON Schema 规范：https://json-schema.org/

Markdown 与代码自检清单（已完成）

1. 标题层级连续：# → ## → ###，未出现跳级。
2. 列表、表格、代码块闭合正确，代码块均包含语言标签（json / ts / python / bash / text）。
3. 所有命令均附用途说明与风险提示（如 0.0.0.0 暴露范围说明）。
4. TypeScript 与 Python 代码语法结构完整，可直接复制到对应文件再细化业务字段。
5. 协议示例、字段约定、测试用例三部分口径一致（trace_id/msg_id/ts_ms 命名统一）。